package com.gyj.zookeeper.Hot.Util;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.zookeeper.*;
import org.apache.zookeeper.data.Stat;

import java.io.IOException;
import java.util.List;

/**
 * Created by Gao on 2018/10/23.
 */
public class ZookeeperUtil {

    private static final Log LOG = LogFactory.getLog(ZookeeperUtil.class);
    private static final int TIMEOUT = 180000;
    private static ZooKeeper zk = null;

    /**
     * listen event type ,No listen event
     */
    public static final int NONE = -1;

    /**
     * listen event type,node create ,delete, data change
     */
    public static final int OTHER = 1;

    /**
     * listen event type, child change(child create and child delete) event an node delete
     */
    public static final int NODECHILDRENCHANGED = 4;

    public static final int NODE_DELETE = 2;

    /**
     * node type, persistent
     */
    public static final int PERSISTENT = 0;

    /**
     * node type ,ephemeral
     */
    public static final int EPHEMERAL = 1;

    /**
     * node type,persistent+sequential
     */
    public static final int PERSISTENT_SEQUENTIAL = 2;

    /**
     * node type,ephemeral+sequential
     */
    public static final int EPHEMERAL_SEQUENTIAL = 3;

    //构造函数初始化的时候就取得zk连接
    public ZookeeperUtil() {
        getConnection();
    }

    public static ZooKeeper getConnection() {
        if (zk == null) {
            initZkConnection();
            LOG.info("init zk.");
        }
        return zk;
    }

    /**
     * 创建节点
     *
     * @param nodePath
     * @param data
     * @param nodeType
     * @throws Exception
     */
    public boolean createNode(String nodePath, String data, int nodeType) throws Exception {
        String path = null;
        try {
            if (nodeType == PERSISTENT) {
                path = zk.create(nodePath, Bytes.toBytes(data), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            } else if (nodeType == EPHEMERAL) {
                path = zk.create(nodePath, Bytes.toBytes(data), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
            } else if (nodeType == PERSISTENT_SEQUENTIAL) {
                path = zk.create(nodePath, Bytes.toBytes(data), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT_SEQUENTIAL);
            } else if (nodeType == EPHEMERAL_SEQUENTIAL) {
                path = zk.create(nodePath, Bytes.toBytes(data), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            } else {
                LOG.error("data type error.");
            }
        } catch (KeeperException e) {
            if (e.code().name().equalsIgnoreCase("NODEEXISTS")) {
                LOG.error(e.getMessage());
            } else {
                throw e;
            }
        }
        if (path != null) {
            LOG.info("create " + nodeType + " node:" + nodePath);
            return true;
        } else {
            return false;
        }
    }

    /**
     * 删除节点
     *
     * @param nodePath
     * @throws Exception
     */
    public void deleteNode(String nodePath) throws Exception {
        try {
            zk.delete(nodePath, -1);
        } catch (Exception e) {
            if (LOG.isErrorEnabled()) {
                LOG.error(e.getMessage(), e);
            }
            throw e;
        }
    }

    /**
     * 得到节点的孩子节点
     *
     * @param nodePath
     * @throws Exception
     */
    public List<String> getChildren(String nodePath) throws Exception {
        List<String> list = zk.getChildren(nodePath, true);
        return list;
    }

    /**
     * 获取节点的数据
     *
     * @param nodePath
     * @return
     * @throws Exception
     */
    public synchronized String getData(String nodePath) throws Exception {
        String content = Bytes.toString(zk.getData(nodePath, false, null));
        return content;
    }

    /**
     * 初始化zk连接
     */
    public static void initZkConnection() {
        //默认读取名称为hbase-site.xml的配置文件
        Configuration conf = HBaseConfiguration.create();
        String IpPort = "";
        String[] str = conf.get("hbase.zookeeper.quorum").split(",");
        for (int i = 0; i < str.length - 1; i++) {
            IpPort += str[i].trim() + ":2181,";
        }
        IpPort += str[str.length - 1].trim() + ":2181";   //127.0.0.1:2181,127.0.0.2:2181,127.0.0.3:2181
        try {
            zk = new ZooKeeper(IpPort, TIMEOUT, getWatcher());
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    private static Watcher getWatcher() {
        return new Watcher() {
            @Override
            public void process(WatchedEvent event) {

            }
        };
    }

    /**
     * 关闭zookeeper连接，释放资源
     */
    public static void close() {
        System.out.println("[SUC-CORE] close");
        if (zk != null) {
            try {
                zk.close();
                zk = null;
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 判断节点是否存在
     *
     * @param nodePath
     * @throws Exception
     */
    public boolean exists(String nodePath) throws Exception {
        Stat stat = zk.exists(nodePath, false);
        if (stat != null) {
            return true;
        } else {
            return false;
        }
    }
}
